package com.atguigu.flink.streamapi.exec;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.pojo.ChannelData;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

/**
 * Created by Smexy on 2022/11/22
 */
public class MySource implements ParallelSourceFunction<String>
{

        boolean ifContinue = true;
        //运行后，源源不断产生数据
        @Override
        public void run(SourceContext<String> ctx) throws Exception {

            String [] behaviors = {"install","uninstall","download","update"};
            String [] channels = {"ZTE","APPLE","HUAWEI","MI","VIVO","OPPO"};

            while (ifContinue){

                ChannelData channelData = new ChannelData(
                    RandomUtils.nextInt(1, 101),
                    behaviors[RandomUtils.nextInt(0, behaviors.length)],
                    channels[RandomUtils.nextInt(0, channels.length)],
                    System.currentTimeMillis()
                );

                //转为json
                String json = JSON.toJSONString(channelData);

                //输出
                ctx.collect(json);

                //每个500ms产生一条
                Thread.sleep(500);

            }

        }

        //调用此方法停止读取数据
        @Override
        public void cancel() {

            ifContinue = false;
        }

}
